package org.pupil.rpc.consumer;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.pupil.rpc.common.ServiceMeta;
import org.pupil.rpc.protocol.RpcProtocol;
import org.pupil.rpc.protocol.RpcRequest;
import org.pupil.rpc.protocol.codec.RpcDecoder;
import org.pupil.rpc.protocol.codec.RpcEncoder;
import org.pupil.rpc.protocol.handler.consumer.RpcResponseHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 消费端
 */
public class RpcConsumer {

    //final 确保只被赋值一次
    private final Bootstrap bootstrap;
    private final EventLoopGroup eventLoopGroup;
    private Logger logger = LoggerFactory.getLogger(RpcConsumer.class);

    public RpcConsumer() {
        bootstrap = new Bootstrap();
        eventLoopGroup = new NioEventLoopGroup(4);
        bootstrap.group(eventLoopGroup)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.SO_KEEPALIVE,true)
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel channel) throws Exception {
                        channel.pipeline()
                                .addLast(new RpcEncoder())
                                .addLast(new RpcDecoder())
                                .addLast(new RpcResponseHandler());
                    }
                });
    }


    /**
     * 发送消息
     * @param protocol 消息
     * @param serviceMeta 服务
     * @throws Exception
     */
    public void sendRequest(RpcProtocol<RpcRequest> protocol, ServiceMeta serviceMeta) throws Exception {
       if (serviceMeta != null){
           ChannelFuture future = bootstrap.connect(serviceMeta.getServiceAddr(), serviceMeta.getServicePort()).sync();
           // 进行异步操作，并添加监听器(ChannelFutureListener)
           future.addListener((ChannelFutureListener) arg0->{
               if (future.isSuccess()){
                   logger.info("连接 rpc server {} 端口 {} 成功.", serviceMeta.getServiceAddr(), serviceMeta.getServicePort());
               }else {
                   logger.error("连接 rpc server {} 端口 {} 失败.", serviceMeta.getServiceAddr(), serviceMeta.getServicePort());
                   future.cause().printStackTrace();
                   eventLoopGroup.shutdownGracefully();
               }
           });
           // 写入数据，发送消息
           future.channel().writeAndFlush(protocol);
       }

    }
}
